package com.galeno.blink.demo;

import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import redis.clients.jedis.Jedis;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/1920:45
 */
public class ProjectDemo {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",8081);
        StreamExecutionEnvironment webUI = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        DataStreamSource<String> sds = webUI.socketTextStream("192.168.77.101", 8888);

        SingleOutputStreamOperator<Tuple3<String, String, Integer>> map = sds.map(x -> {
            String[] split = x.split(",");
            return Tuple3.of(split[0], split[1], Integer.parseInt(split[2]));
        }).returns(TypeInformation.of(new TypeHint<Tuple3<String, String, Integer>>() {
        }));
        SingleOutputStreamOperator<Tuple> project = map.project(2, 1, 0);
        project.print();
        webUI.execute();


    }
}
